package net.wicp.tams.common.binlog.alone.checkpoint;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.binlog.alone.BusiAssit;
import net.wicp.tams.common.binlog.alone.ListenerConf.ColHis;
import net.wicp.tams.common.binlog.alone.ListenerConf.Position;
import net.wicp.tams.common.binlog.alone.normalize.ISaveCheckPoint;
import net.wicp.tams.common.constant.dic.YesOrNo;

@Slf4j
public class CheckPointMemory implements ISaveCheckPoint {

	// 注意清理此list，长期运行会导致数量大，占用内存
	private final List<PositionPack> posList = new ArrayList<PositionPack>();

	private final int maxPosSize = 100;

	private final Map<String, List<ColHis>> colMap = new HashMap<String, List<ColHis>>();

	@Override
	public void init(net.wicp.tams.common.binlog.alone.ListenerConf.ConnConf.Builder connConfBuilder) {
	}

	@Override
	public void shutdown() {
		posList.clear();
		colMap.clear();
	}

	@Override
	public void savePoint(Position pos) {
		PositionPack positionPack = new PositionPack(pos);
		if (!posList.contains(positionPack)) {
			posList.add(positionPack);
			if (posList.size() > maxPosSize) {// 最多存100个位点，防止无限增长
				PositionPack l1 = posList.get(posList.size() - 1);
				PositionPack l2 = posList.get(posList.size() - 2);
				PositionPack l3 = posList.get(posList.size() - 3);
				posList.clear();
				posList.add(l3);
				posList.add(l2);
				posList.add(l1);
			}
		}
		log.info("保存实例:{} 位点{}成功",pos.getServerIp(),pos.getGtids());
	}

	@Override
	public void saveColName(ColHis colHis) {
		log.info("the newer colHis is:{}", colHis);
		String key = BusiAssit.getColHiskey(colHis.getDb(), colHis.getTb());
		if (colMap.containsKey(key)) {
			List<ColHis> templist = colMap.get(key);
			templist.add(colHis); // 倒序
			Collections.sort(templist, new Comparator<ColHis>() {
				@Override
				public int compare(ColHis o1, ColHis o2) {
					long def = o2.getTime() - o1.getTime();
					return def > 0 ? 1 : (def == 0 ? 0 : -1);
				}
			});
			colMap.put(key, templist);

		} else {
			ArrayList<ColHis> templist = new ArrayList<ColHis>();
			templist.add(colHis);
			colMap.put(key, templist);
		}
	}

	@Override
	public Position findPoint(long time) {
		for (int i = posList.size(); i > 0; i--) {
			CheckPointMemory.PositionPack positionPack = posList.get(i);
			if (positionPack.getPos().getTime() <= time) {
				return positionPack.getPos();
			}
		}
		return null;
	}

	@Override
	public List<ColHis> findColsList(String db, String tb) {
		String key = BusiAssit.getColHiskey(db, tb);
		if (CollectionUtils.isEmpty(colMap.get(key))) {
			return new ArrayList<>();
		}
		return colMap.get(key);
	}

	@Override
	public List<ColHis> findColsAll() {
		List<ColHis> allColHis = new ArrayList<>();
		for (String key : colMap.keySet()) {
			allColHis.addAll(colMap.get(key));
		}
		return allColHis;
	}

	@Data
	private class PositionPack {
		private final Position pos;

		public PositionPack(Position pos) {
			this.pos = pos;
			this.gtids = pos.getGtids();
		}

		private String gtids;

		@Override
		public boolean equals(Object PositionPack) {
			PositionPack temp = (PositionPack) PositionPack;
			return this.pos.getGtids().equals(temp.getGtids());

		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result + ((gtids == null) ? 0 : gtids.hashCode());
			return result;
		}
	}

	// 不支持分布式锁
	@Override
	public YesOrNo acquireLock() {
		return YesOrNo.yes;
	}

	@Override
	public void releaseLock() {

	}

}
